package com.atguigu.Flink.sql.time;

import com.atguigu.Flink.POJO.Event;
import com.atguigu.Flink.function.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class Flink02_EventTime {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<Event> ds = env.addSource(new ClickSource()).assignTimestampsAndWatermarks(
                WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event event, long l) {
                                        return event.getTs();
                                    }
                                }
                        )
        );
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //将流转换为表，并指定处理时间字段
        Schema schema = Schema.newBuilder()
                .column("user", "STRING")
                .column("url" , "STRING")
                .column("ts" , "BIGINT")
                //指定处理时间字段
                .columnByExpression("pt" , "proctime()")
                //指定事件时间字段
                .columnByExpression("et" , "TO_TIMESTAMP_LTZ(ts, 3) ")
                //指定水位线的生成
                //1.如果流中已经生成过水位线， 可以直接沿用流中的水位线
                //.watermark("et" , "source_watermark()")
                //2. 也可以重新指定
                .watermark("et" ,  "et - INTERVAL '0' SECOND ")
                .build();
        Table table = tableEnv.fromDataStream(ds, schema);
        table.printSchema();

        //Sql中声明，连接器中声明
        String createTable =
                " create table t1 (" +
                        " id STRING , " +
                        " vc INT, " +
                        " ts BIGINT , " +
                        " pt AS proctime() , " +
                        " et AS TO_TIMESTAMP_LTZ(ts ,0 ) , " + // 先基于当前数据中的时间，转换成Flink认识的TimeStamp或者TimeStamp_Ltz类型。
                        " WATERMARK FOR et AS et - INTERVAL '0.5' SECOND " +
                        " ) with (" +
                        " 'connector' = 'print' " +
                        ")" ;
        tableEnv.executeSql(createTable);
        tableEnv.from("t1").printSchema();



        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
